package com.tech.common

import org.apache.zookeeper.{WatchedEvent, Watcher, ZooKeeper}

import scala.collection.mutable

object ZkUtil {

  def main(args: Array[String]): Unit = {

    println(getOffset("retailevent_uat_sz","dev-cdp-03:2181,dev-cdp-04:2181"))

  }

  def getOffset(topics:String,zkMaster:String): String ={

    val map = KuduUtil.getOffset()

    var result = "{"

    topics.split(",").foreach(
      topic => {
        result += getOffsetFromZk(topic,map,zkMaster)
        result += ","
      }
    )

    result = result.substring(0,result.length-1)
    result += "}"

    result

  }

  def getOffsetFromZk(topic:String,map:mutable.HashMap[String,Long],zkMaster:String): String ={

    val zooKeeper = new ZooKeeper(zkMaster, 1000, new Watcher {
      override def process(watchedEvent: WatchedEvent): Unit = {}
    })

    // zookeeper中保存partitions的路径
    val partitions = s"/brokers/topics/$topic/partitions"


    val strings = zooKeeper.getChildren(partitions, true)

    val iterator = strings.iterator()

    var i:String = "\"" + topic + "\":{"

    while (iterator.hasNext){
      val partition = iterator.next()
      val offset = if (map.contains(topic + "-" + partition)) {
        map(topic + "-" + partition)
      } else {
        -2
      }
      i += "\"" + partition + "\":" + offset + ","

    }

    i = i.substring(0,i.length-1)
    i += "}"

    i
  }


}
